package com.ttx.zookeeper.cluster;

import com.ttx.zookeeper.DefaultWatcher;
import org.apache.zookeeper.*;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.zookeeper.Watcher.Event.EventType.NodeChildrenChanged;
import static org.apache.zookeeper.Watcher.Event.EventType.NodeCreated;
import static org.apache.zookeeper.Watcher.Event.EventType.NodeDataChanged;

/**
 * 集群管理,
 *  实现高可用, master执行任务, slave 等待, master宕机后, slave变成master, 继续执行任务
 *
 *  leader选举
 *
 *
 *  官方文档
 *  https://zookeeper.apache.org/doc/r3.5.7/recipes.html
 *
 *
 *  A simple way of doing leader election with ZooKeeper is to use the SEQUENCE|EPHEMERAL flags
 *  when creating znodes that represent "proposals" of clients.
 *
 *
 *
 *  1. Create znode z with path "ELECTION/guid-n_" with both SEQUENCE and EPHEMERAL flags;
 *  2. Let C be the children of "ELECTION", and i be the sequence number of z;
 *  3. Watch for changes on "ELECTION/guid-n_j", where j is the largest sequence number such that j < i and n_j is a znode in C;
 *
 * @author TimFruit
 * @date 20-2-24 下午9:19
 */
public class RawWorker {

    private static final String clusterPath="/cluster";
    private static final String nodesPath=clusterPath+"/nodes";
    private static final String confPath=clusterPath+"/conf";





    public static void main(String[] args) throws Exception {

        String connectString="127.0.0.1:2181";

        int timeout=(int)MILLISECONDS.convert(10, SECONDS);


        final ZooKeeper zk=new ZooKeeper(connectString, timeout, new DefaultWatcher(),null);


        if(zk.exists(clusterPath, false)==null){
               zk.create(clusterPath, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        if(zk.exists(nodesPath, false)==null){
                   zk.create(nodesPath, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

        


        //1. 将自身注册到集群节点中
        final String guid=UUID.randomUUID().toString();
        System.out.println("当前workId: "+guid);


        final String split="_";

        // 创建临时且有序列号的节点
        String workerPath=nodesPath+"/"+guid+split;
        zk.create(workerPath, guid.getBytes(),  ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);









        //2. 如果是master,则继续执行, 否则等待
        List<String> nodes=zk.getChildren(nodesPath, false);
        //注意事项: 返回的节点名字并没有根据sequence排序
        // nodeName= guid+spilt+sequence
        sortedBySequence(nodes, split, 1);


        String firstNode=nodes.get(0);
        System.out.println("firstNode: "+firstNode);

        String[] strs=firstNode.split(split);

        //2.1. 获取的是第一个节点
        if(strs[0].equals(guid)){
            System.out.println("本work成为master节点, 将进行工作....");


        } else{

            //2.2. 非第一个节点， 等待并监听
            System.out.println("本work成为slave节点, 将Stanby...");




            final CountDownLatch cdl=new CountDownLatch(1);

            zk.getChildren(nodesPath, new Watcher() {
                public void process(WatchedEvent watchedEvent) {
                    String path=watchedEvent.getPath();

                    System.out.println(path);

                    if(!path.startsWith(nodesPath)){
                        System.out.println("不是cluster事件");
                        return;
                    }

                    Event.EventType et=watchedEvent.getType();
                    if(et== NodeDataChanged
                            || et == NodeChildrenChanged
                            || et == NodeCreated){

                        List<String> nodes= null;
                        try {
                            nodes = zk.getChildren(nodesPath, false);
                        } catch (KeeperException e) {
                            e.printStackTrace();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        String firstNode=nodes.get(0);
                        System.out.println("firstNode: "+firstNode);

                        String[] strs=firstNode.split("_");
                        if(strs[0].equals(guid)) {
                            System.out.println("本work成为master节点, 将进行工作....");
                            cdl.countDown();
                        }
                    }


                }
            });
            cdl.await();
        }








        //3. 可以从zookeeper读取配置, 或者初始化配置
        // （即， 使得本程序无状态， 将状态(元数据)保存在zookeeper， 本程序可以扩展）



        //4. 执行任务
         work();







    }




    public static void work() throws InterruptedException {
        System.out.println("假装工作(等待)...");
        Thread.sleep(Long.MAX_VALUE);
    }



    protected  static void sortedBySequence(List<String> children, String spilt, int seqIndex){
        // child, seq
        Map<String, String> childSequenceMap=children.stream().collect(Collectors.toMap(
                s-> s,
                s -> {
                    // guid+spilt+Sequence;
                    String[] strs=s.split(spilt);
                    String seq=strs[seqIndex];
                    return seq;
                }
        ));

        Collections.sort(children, (s1, s2)->{
            String seq1=childSequenceMap.get(s1);
            String seq2=childSequenceMap.get(s2);
            return seq1.compareTo(seq2);
        });

    }



}
